package com.example.day04.controller.sideoutputs;

import com.example.day04.controller.bean.WaterSemsor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * nc -lk 7777
 */
public class WaterDemo1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("node3", 7777);

        SingleOutputStreamOperator<WaterEntity> map = source.map((String s) -> {
            String[] split = s.split(",");
            return new WaterEntity(split[0], Integer.parseInt(split[1]), split[2]);
        });

        WatermarkStrategy<WaterEntity> waterSemsorWatermarkStrategy = WatermarkStrategy.<WaterEntity>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, recordTimestamp) -> element.getVc() * 1000L);


        SingleOutputStreamOperator<WaterEntity> sensorDSwithWatermark = map.assignTimestampsAndWatermarks(waterSemsorWatermarkStrategy);

        SingleOutputStreamOperator<String> process = sensorDSwithWatermark.keyBy(WaterEntity::getId).process(new KeyedProcessFunction<String, WaterEntity, String>() {
            @Override
            public void processElement(WaterEntity value, KeyedProcessFunction<String, WaterEntity, String>.Context ctx, Collector<String> out) throws Exception {
                Long currenttimestamp = ctx.timestamp();

                TimerService timerService = ctx.timerService();
                timerService.registerEventTimeTimer(5000L);

                System.out.println("当前的时间是： " + currenttimestamp + "注册了一个5s的定时器");
//                timerService.registerProcessingTimeTimer(5000L);
//                timerService.deleteEventTimeTimer(5000L);
//                timerService.deleteProcessingTimeTimer(5000L);
//                timerService.currentProcessingTime();
//                timerService.currentWatermark();
                out.collect(value.getId());
            }

            /**
             * 当前时间推进到定时器的注册时间，就会触发执行
             * @param timestamp 当前时间进展
             *                    上下文
             *                  采集器
             *
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterEntity, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                System.out.println("现在时间是" + timestamp + "定时机触发了");
            }
        });
        process.print();
        env.execute();
    }
}
